package cn._51doit.day07;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashSet;

/**
 * @create: 2021-10-23 19:55
 * @author: 今晚打脑斧先森
 * @program: AdCount
 * @Description: 统计广告曝光\点击的次数和人数
 * 数据
 * gg,user1,click   click点击
 * gg,user1,view    view浏览
 * 输出 gg,click,人数,view,次数
 **/
public class ValueStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(10000);

        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Tuple3<String, String, String>> tpStream = lines.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(split[0], split[1], split[2]);
            }
        });

        //对数据进行keyby
        KeyedStream<Tuple3<String, String, String>, Tuple2<String, String>> keyedStream = tpStream.keyBy(new KeySelector<Tuple3<String, String, String>, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(Tuple3<String, String, String> value) throws Exception {
                return Tuple2.of(value.f0, value.f2);
            }
        });

        SingleOutputStreamOperator<Tuple4<String, String, Integer, Integer>> res = keyedStream.map(new RichMapFunction<Tuple3<String, String, String>, Tuple4<String, String, Integer, Integer>>() {
            private ValueState<Integer> countState;
            private ValueState<HashSet<String>> uidState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //定义一个状态描述器
                //统计浏览次数
                ValueStateDescriptor<Integer> countStateDescriptor = new ValueStateDescriptor<>("浏览次数", Integer.class);
                countState = getRuntimeContext().getState(countStateDescriptor);
                //定义集合统计人数
                ValueStateDescriptor<HashSet<String>> hashSetValueStateDescriptor = new ValueStateDescriptor<>("用户id", TypeInformation.of(new TypeHint<HashSet<String>>() {
                }));
                uidState = getRuntimeContext().getState(hashSetValueStateDescriptor);
            }

            @Override
            public Tuple4<String, String, Integer, Integer> map(Tuple3<String, String, String> value) throws Exception {
                String gg = value.f0;
                String beHavior = value.f1;//行为
                String uid = value.f2;

                //对浏览人名进行统计
                HashSet<String> hashSet = uidState.value();
                if (hashSet==null) {
                    hashSet = new HashSet<>();
                }
                hashSet.add(uid);
                uidState.update(hashSet);

                //对浏览人数进行统计
                Integer historyCount = countState.value();
                if (historyCount == null) {
                    historyCount = 0;
                }
                historyCount += 1;
                countState.update(historyCount);

                return Tuple4.of(gg, beHavior, historyCount, hashSet.size()); //广告 用户 浏览的次数  点击的次数
            }
        });
        res.print();
        env.execute();
    }
}
